package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.event.model.broker.PushDelayMsgEvent;
import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.store.TimeWheelModelManager;
import cn.me.alphamq.broker.util.QueueIdUtil;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushDelayMsgReq;
import cn.me.alphamq.common.entity.resp.PushDelayMsgResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.filter.SimplePropertyPreFilter;
import io.netty.channel.ChannelHandlerContext;

/**
 * 生产者生产延迟消息的事件的监听器
 *
 * @author f
 */
public class PushDelayMsgListener implements Listener<PushDelayMsgEvent> {

    /**
     * 避免将 pushDelayMsgReq 对象的消息 id 序列化到文件中的过滤器
     */
    private final SimplePropertyPreFilter simplePropertyPreFilter = new SimplePropertyPreFilter();

    {
        simplePropertyPreFilter.getExcludes().add("msgId");
    }

    @Override
    public void onReceive(PushDelayMsgEvent event) throws Exception {
        PushDelayMsgReq pushDelayMsgReq = event.getDelayMsgReq();
        ChannelHandlerContext ctx = event.getCtx();
        String msgId = event.getMsgId();

        // 将其投放到时间轮中
        pushDelayMsgReq.setQueueId(QueueIdUtil.getQueueId(pushDelayMsgReq.getTopic(), pushDelayMsgReq.getQueueId()));
        TimeWheelModelManager.add(DelayMessageModel.newDelayMsgModel(pushDelayMsgReq));

        // 将消息持久化到延迟队列中
        Container.getCommitLogAppendHandler().appendMsg(Message.newDelayMsg(
                BrokerConstant.DELAY_TOPIC_NAME,
                Container.getBrokerProperties().getBrokerCurrDelayQueueId(),
                JSON.toJSONBytes(pushDelayMsgReq, simplePropertyPreFilter)
        ));

        if (msgId != null) {
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_DELAY_MSG_RESP,
                    JSON.toJSONBytes(new PushDelayMsgResp(msgId))));
        }
    }
}
